<?php

namespace App\Task;

use App\Model\Live;
use App\Model\UserModelAnchor;
use App\Utils\Enum\ImEnum;
use App\Utils\Enum\LiveEnum;
use App\Utils\Im;
use App\Websocket\ImWebsocket;
use Exception;
use Hyperf\Redis\Redis;
use Hyperf\WebSocketServer\Sender;
use Psr\Container\ContainerExceptionInterface;
use Psr\Container\NotFoundExceptionInterface;
use RedisException;
use function Hyperf\Support\env;

class LiveMonitorTask extends AbstractBaseTask
{
    public function execute(): void
    {
        //断流超时自动关播
        $this->_checkCuteStreamTimeout();
    }

    /**
     * 断流超时自动关播
     * @return void
     */
    private function _checkCuteStreamTimeout(): void
    {
        //允许最长断流时间
        $maxCuteStreamTime = (int)env('LIVE_MAX_CUTE_STREAM_TIME', 60);
        $liveList = Live::query()->where([['status', '=', LiveEnum::STATUS_LIVING], ['cute_stream_time', '<>', 0], ['cute_stream_time', '<', time() - $maxCuteStreamTime]])->get()->toArray();
        foreach ($liveList as $live) {
            Live::query()->where(['id' => $live['id'], 'status' => LiveEnum::STATUS_LIVING])->update([
                'status' => LiveEnum::STATUS_CLOSED
            ]);
            UserModelAnchor::query()->where('room_id', $live['room_id'])->update(['live_status' => LiveEnum::STATUS_CLOSED]);
            $this->_pushMessage($live['room_id']);
        }
    }

    /**
     * 下发下播消息
     * @param $roomId
     * @return void
     */
    private function _pushMessage($roomId): void
    {
        try {
            $redis = $this->container->get(Redis::class);
            $fds = $redis->sMembers(ImWebsocket::KeyGroup($roomId));
            if (!empty($fds)) {
                $sender = $this->container->get(Sender::class);
                $msg = Im::Create('ImDoResult', ['code' => ImEnum::IM_DO_LIVE_OFF, 'data' => new_json_encode(['roomId' => $roomId, 'status' => LiveEnum::STATUS_CLOSED])]);
                foreach ($fds as $fd) {
                    $fd = (int)$fd;
                    try {
                        if ($sender->check($fd)) {
                            $sender->push($fd, $msg, WEBSOCKET_OPCODE_BINARY);
                        }
                    } catch (Exception $e) {
                        $this->logger->error(sprintf('推送下播消息失败 fd=%d error=%s', $fd, $e->getMessage()));
                    }
                }
            }
        } catch (NotFoundExceptionInterface|ContainerExceptionInterface|RedisException $e) {
            $this->logger->error(sprintf('推送下播消息失败 error=%s', $e->getMessage()));
        }
    }
}